package xw4g08.sameAs;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Random;
import java.util.Set;

import org.apache.commons.math.MathException;
import org.apache.commons.math.distribution.ZipfDistributionImpl;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.ValueFactory;
import org.openrdf.query.BindingSet;
import org.openrdf.query.MalformedQueryException;
import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQuery;
import org.openrdf.query.TupleQueryResult;
import org.openrdf.repository.RepositoryConnection;
import org.openrdf.repository.RepositoryException;
import org.openrdf.rio.RDFFormat;
import org.openrdf.rio.RDFHandlerException;
import org.openrdf.rio.RDFWriter;
import org.openrdf.rio.Rio;

import xw4g08.Configuration;
import xw4g08.StatisticsSPARQL1_1;

public class GeneratorCore {

	//private final double staRate = 3449341.0/2000000000;//the ratio of sameAs statements
	private final double resRate = 4174654.0/2000000000;//the ratio of resources having sameAs
	//private final double a = 
	private final double exp = 2.528;
	
	/**
	 * Generate sameAs statements for given RDF graph. The sameAs statements are written to 
	 * the directory specified by {@code Configuration.OUT_DIR} and {@code outName}.
	 * @param con The repository connection of source data.
	 * @param outName File name of generated sameAs statements.
	 */
	void sameasWriter(RepositoryConnection con, String outName) {
		if(!Configuration.OUT_DIR.exists())
			Configuration.OUT_DIR.mkdir();
		int statementCounter = 0;
		
		File outFile = new File(Configuration.OUT_DIR,"sameAs_"+outName + ".n3");
		RDFFormat outFormat = RDFFormat.N3;
		
		Random ran = new Random();
		try {
			ValueFactory vf = con.getValueFactory();
			URI sameAs = vf.createURI("http://www.w3.org/2002/07/owl#","sameAs");
			List<Resource> allNodes = new ArrayList<Resource>();
			allNodes.addAll(getNodes(con));
			int size = allNodes.size();
			int n = (int) (size*resRate);//the number of nodes having sameAs URIs
			if(n<10)
				n = 10;//if the number is too small, set to 10
			ZipfDistributionImpl zipf = new ZipfDistributionImpl(n, exp);
			FileOutputStream out = new FileOutputStream(outFile);
			RDFWriter writer = Rio.createWriter(outFormat, out);
			writer.startRDF();
			StatisticsSPARQL1_1.handelNS(con, writer);
			writer.handleNamespace("owl", "http://www.w3.org/2002/07/owl#");
			//Randomly select n nodes out of all the nodes as those having sameAs URIs. 
			//For each of them, generate m sameAs URIs of it according to Zipf's law, 
			//and then randomly select m nodes (of the same type of the original node if possible) 
			//to generate sameAs statements.
			for(int i=0;i<n;i++) {
				int subInx = ran.nextInt(size);//index of the subject
				Resource subjectNode = allNodes.get(subInx);
				//////////
				List<Resource> sameTypeN = sameTypeNodes(con,subjectNode);
				if(sameTypeN.isEmpty()) {//if no specific type is presented, use all nodes as potential objects
					sameTypeN.addAll(allNodes);
				}
				int m = nextPow(zipf);
				if(m > sameTypeN.size()) {
					System.out.println("Not enough nodes in the same type of "+ subjectNode+ ". "+ (m-sameTypeN.size())+" more are needed.");
					m = sameTypeN.size();
				}
					
				/////////
				
				for(int j=0;j<m;j++) {
					//if there's only one node and it's the same as the subject, jump to the next subject
					if(sameTypeN.size()==1 && subjectNode.equals(sameTypeN.get(0)))
						continue;
					//find a node distinct from the subject as the co-referencing object
					int objInx;
					while(true) {
						objInx = ran.nextInt(sameTypeN.size());
						if(!subjectNode.equals(sameTypeN.get(objInx)))
						break;
					}
					
					
					Statement sta = vf.createStatement(subjectNode, sameAs, sameTypeN.get(objInx));
					//Statement sta = vf.createStatement(allNodes.get(subInx), sameAs, allNodes.get(objInx));
					writer.handleStatement(sta);
					sta = vf.createStatement(sameTypeN.get(objInx), sameAs, subjectNode);//generate the symmetric statement as well
					//sta = vf.createStatement(allNodes.get(objInx), sameAs, allNodes.get(subInx));
					writer.handleStatement(sta);
					statementCounter++;
				}
			}
			writer.endRDF();
			System.out.println(statementCounter + " statements have been written to " + outFile);
			con.close();
		} catch (RepositoryException e) {
			e.printStackTrace();
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (RDFHandlerException e) {
			e.printStackTrace();
		} catch (MalformedQueryException e) {
			e.printStackTrace();
		} catch (QueryEvaluationException e) {
			e.printStackTrace();
		}
	}
	
	private List<Resource> sameTypeNodes(RepositoryConnection con, Resource node) throws 
	RepositoryException, MalformedQueryException, QueryEvaluationException {
		List<Resource> rel = new ArrayList<Resource>();
		String qStr = "PREFIX rdf:<http://www.w3.org/1999/02/22-rdf-syntax-ns#> " +
						"SELECT DISTINCT ?o " + "{ <"+ node.stringValue() +
						"> rdf:type ?type. ?o rdf:type ?type FILTER (?o != <"+node.stringValue()+">) }";
		TupleQuery qry = con.prepareTupleQuery(QueryLanguage.SPARQL, qStr);
		TupleQueryResult results = qry.evaluate();
		while(results.hasNext()) {
			BindingSet binding = results.next();
			Value o = binding.getValue("o");
			rel.add((Resource)o);
		}
		return rel;
	}
	
	private Set<Resource> getNodes(RepositoryConnection con) throws RepositoryException, 
	MalformedQueryException, QueryEvaluationException {
		Set<Resource> resources = new HashSet<Resource>();
		String qStr = "SELECT DISTINCT ?s ?o {?s ?p ?o}";
		TupleQuery qry = con.prepareTupleQuery(QueryLanguage.SPARQL, qStr);
		TupleQueryResult results = qry.evaluate();
		while(results.hasNext()) {
			BindingSet binding = results.next();
			Value sub = binding.getValue("s");
			if(sub instanceof Resource) {
				resources.add((Resource) sub);
			}
			Value obj = binding.getValue("o");
			if(obj instanceof Resource) {
				resources.add((Resource) obj);
			}
		}
		
		return resources;
	}
	
	public static int nextPow(ZipfDistributionImpl zipf) {
		int x = 0;
		try {
			x = zipf.sample();
		} catch (MathException e) {
			e.printStackTrace();
		}
		return x;
	}

	
	
	
}
